package com.bigdata.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession, expressions}

object Demo8Burks {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("burk")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    //读取数据
    val burksDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING,year STRING,tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .load("data/burks.txt")

    // burksDF.show()

    /**
     * sql
     * 1、统计每个公司每年按月累计收入  行转列 --> sum窗口函数
     * 输出结果
     * 公司代码,年度,月份,当月用电量,累计用电量
     */
    burksDF.createOrReplaceTempView("burks")

    // 行转列，原来一行12列数据，转为12行数据，每一行的数据要包含公司标识、年、月、本月用电量、本年的总用电量
    spark.sql(
      """
        |select t1.*, sum(sal) over(partition by burk,year order by month asc) as sum_sal from (
        |select burk, year, month, sal from burks
        |lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,
        |tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) temp as month,sal
        |) t1
        |""".stripMargin)
    // .show()

    /**
     * DSL
     * expr("1") 表示的是常量1
     * $"tsl01" 表示的是引用表中的字段
     */
    val columns: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )

    burksDF
      .select($"burk", $"year", explode(columns) as Array("month", "sal"))
      // 开窗得到每年的总用电量
      .withColumn("sum_sal", sum($"sal") over Window.partitionBy($"burk", $"year").orderBy($"month".asc))
    // .show()

    /**
     * 2、统计每个公司当月比上年同期增长率  行转列 --> lag窗口函数， 同比增长率
     * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
     * 同比（同比增长率）是以上年同期为基期相比较，即本期某一时间段与上年某一时间段相比，如2018年4月份与2017年4月份相比较。
     * 　计算公式为：同比增长率 ＝（本期数－同期数）／ 同期数 × 100%
     * 环比（环比增长率）是与上一个相邻统计周期相比较，即n月与第n-1月的比较。 如2018年4月份与2018年3月份相比较。
     * 计算公式为：环比增长率 =（本期数－上期数）／上期数 × 100%
     * 环比侧重反映数据的短期变化，同比则侧重反映长期趋势。
     */
    burksDF
      .select($"burk", $"year", explode(columns) as Array("month", "sal"))
      .withColumn("tong_last_sal", lag($"sal", 1, 0)
        over Window.partitionBy($"burk", $"year").orderBy($"month".asc))
      .withColumn("tong_grow", ($"sal" - $"tong_last_sal") / $"tong_last_sal")
      //      .withColumn("last_sal", lag($"sal", 1) over Window.partitionBy($"burk", $"month").orderBy($"year"))
      //      .withColumn("grow", round($"sal" / $"last_sal" - 1, 2))
      //      // .withColumn("grow1", round(($"sal" - $"last_sal") / $"last_sal", 2))
      //      .withColumn("process_grow", when($"grow".isNull, 0).otherwise($"grow"))
      //      .select($"burk", $"year", $"month", $"process_grow")
      .show(36)

  }

}